package com.swsc.ai.tasks;

import com.swsc.ai.enums.TFEnum;
import com.swsc.ai.taskfactory.Task;
import com.swsc.ai.util.HDFSUtil;
import com.swsc.ai.util.SparkUtil;
import org.apache.spark.sql.SparkSession;

import java.sql.SQLException;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Map;

/**
 * @describe:
 * @author: 容若
 * @created: 2023-12-29 10:12
 */
public abstract class ForecastTempTask implements Task {
    @Override
    public void run(Map<String, String> map) throws Exception {
        String taskName = map.getOrDefault("taskName", "test");
        String projectName = map.getOrDefault("projectName", "ai_marketing");
        String className = map.getOrDefault("className", "test");
        String subTaskName = map.getOrDefault("subTaskName", "test");
        String dt = map.getOrDefault("dt", LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")));
        String isTest = map.getOrDefault("isTest", TFEnum.FALSE.getName());
        /**
         * 获取画像宽表全路径
         */
        String portraitPath = map.getOrDefault("portraitPath", "test");

        /**
         * 获取产品购买可能性预测全路径
         */
        String forecastPath = map.getOrDefault("forecast", "test");

        /**
         * 获取产品购买可能性预测全路径
         */
        String kmeansResultPath = map.getOrDefault("kmeansResult", "test");

        String fileName = HDFSUtil.getOutHDFSPath(projectName, taskName, subTaskName, dt);
        if (TFEnum.TRUE.equals(TFEnum.fromTypeName(isTest))) {
            fileName = HDFSUtil.getTestOutHDFSPath(className, subTaskName, dt);
        }
        SparkSession session = SparkUtil.getSession(isTest, className);
        createView(session, portraitPath, forecastPath, kmeansResultPath);
        calculateEstimationResult(session, fileName);
        if (null != session){
            session.stop();
        }
    }

    public abstract void createView(SparkSession session, String... str) throws SQLException;
    public abstract void calculateEstimationResult(SparkSession session, String... str);
}
